package implementation.client.net;

import java.net.InetSocketAddress;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executors;

import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
import org.jboss.netty.handler.codec.serialization.ObjectDecoder;
import org.jboss.netty.handler.codec.serialization.ObjectEncoder;

import implementation.client.IClientConnectionManager;
import implementation.shared.IPacket;

public class ClientConnectionManager implements IClientConnectionManager {
	
	private final ConcurrentLinkedQueue<IPacket> _inbox;
	private final ClientBootstrap _bootstrap;
	private Channel _channel;
	private final long TIMEOUT = 2000; 
	
	public ClientConnectionManager() {
		
		this._inbox = new ConcurrentLinkedQueue<IPacket>();
		
		this._bootstrap = new ClientBootstrap(new NioClientSocketChannelFactory(
				Executors.newCachedThreadPool()
				, Executors.newCachedThreadPool()));
		
		this._bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
			
			@Override
			public ChannelPipeline getPipeline() throws Exception {
				ChannelPipeline pipeline = Channels.pipeline();
				pipeline.addLast("objectEncoder", new ObjectEncoder());
				pipeline.addLast("stringDecoder", new ObjectDecoder());
				pipeline.addLast("handler", new ClientChannelHandler(_inbox));
				return pipeline;
				
			}
		});
		
		this._bootstrap.setOption("tcpNoDelay", true);
		this._bootstrap.setOption("keepAlive", true);
		
	}

	@Override
	public boolean connect(String host, int port) {
		
		if (this._channel != null && this._channel.isConnected()) {
			return true;
		}
		ChannelFuture future = this._bootstrap.connect(new InetSocketAddress(host, port));
		future.awaitUninterruptibly(TIMEOUT);
		this._channel = future.getChannel();
		future.awaitUninterruptibly();
		return future.isSuccess();
	}

	@Override
	public boolean send(IPacket packet) {
		//String str = this._serializer.serialize(packet);
		//System.out.println("serialized:"+str);
		ChannelFuture future = this._channel.write(packet);
		future.awaitUninterruptibly(TIMEOUT);
		return future.isSuccess();
	}

	@Override
	public IPacket recieve() throws InterruptedException {
		IPacket msg = null;
		synchronized (this._inbox) {
			while (this._inbox.size() == 0) {
				this._inbox.wait();
			}
			msg = this._inbox.poll();
		}
		return msg;
	}

	@Override
	public void close() {
		ChannelFuture future = this._channel.close();
		future.awaitUninterruptibly(TIMEOUT);
		this._channel.unbind().awaitUninterruptibly(TIMEOUT);
		this._channel.close().awaitUninterruptibly(TIMEOUT);
		this._bootstrap.releaseExternalResources();
	}

}
